package com.leal.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 简单java api
 */
public class HbaseClient {

    private static final Configuration conf = HBaseConfiguration.create();
    private static Admin admin;
    public static Connection connection = null;

    static {
        //HBase Master地址，在hbase-site.xml中配置
        conf.set("hbase.rootdir", "hdfs://127.0.0.1:9000/hbase");
        //ZooKeeper三个服务器地址
        conf.set("hbase.zookeeper.quorum", "127.0.0.1");

        try {
            connection = ConnectionFactory.createConnection(conf);
            // 非线程安全
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void closeConnection(Connection connection) throws IOException {
        if (connection != null) {
            connection.close();
        }
    }

    // DDL

    public static void createNamespace(String namespace) throws IOException {

        // 创建明明空间的描述
        NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(namespace);
        builder.addConfiguration("user", "leal");

        // 调用build 完成构建
        try {
            admin.createNamespace(builder.build());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // close
        admin.close();
    }

    public static boolean isTableExists(String namespace, String tableName) {
        try {
            return admin.tableExists(TableName.valueOf(namespace, tableName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * create table by col family
     *
     * @param tableName    name
     * @param columnFamily columnFamily
     */
    public static void createTable(String tableName, String... columnFamily) {

        TableName tableNameObj = TableName.valueOf(tableName);
        try {
            if (admin.tableExists(tableNameObj)) {
                System.out.println("Table: " + tableName + " already exists!");
            } else {
                HTableDescriptor tb = new HTableDescriptor(tableNameObj);
                for (String s : columnFamily) {
                    HColumnDescriptor family = new HColumnDescriptor(s);
                    tb.addFamily(family);
                }
                admin.createTable(tb);
                System.out.println(tableName + "创建表成功");
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println(tableName + "创建表失败");
        }
    }

    public static void addData(String tableName) throws IOException {

        //获取表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put("0001".getBytes());
        put.addColumn("info".getBytes(), "id".getBytes(), Bytes.toBytes(1));
        put.addColumn("info".getBytes(), "name".getBytes(), Bytes.toBytes("小明"));
        put.addColumn("info".getBytes(), "age".getBytes(), Bytes.toBytes(22));
        put.addColumn("info".getBytes(), "sex".getBytes(), Bytes.toBytes("男"));
        put.addColumn("info".getBytes(), "address".getBytes(), Bytes.toBytes("青岛"));
        put.addColumn("info".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000001"));
        put.addColumn("info".getBytes(), "say".getBytes(), Bytes.toBytes("你好"));
        table.put(put);
        //关闭表
        table.close();
        System.out.println("插入数据成功");
    }

    public static void insertBatchData(String tableName) throws IOException {

        //获取表
        Table table = connection.getTable(TableName.valueOf(tableName));
        //创建put对象，并指定 row key
        Put put = new Put("0002".getBytes());
        put.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(1));
        put.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小红"));
        put.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(21));
        put.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("女"));
        put.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("济南"));
        put.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000002"));
        put.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("Hello！"));

        Put put2 = new Put("0003".getBytes());
        put2.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(2));
        put2.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小强"));
        put2.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(22));
        put2.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("男"));
        put2.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("北京"));
        put2.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000003"));
        put2.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("俺北京……"));

        Put put3 = new Put("0004".getBytes());
        put3.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(4));
        put3.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小花"));
        put3.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(23));
        put3.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("女"));
        put3.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("重庆"));
        put3.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000004"));
        put3.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("要得"));

        Put put4 = new Put("0005".getBytes());
        put4.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(5));
        put4.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小杰"));
        put4.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(24));
        put4.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("男"));
        put4.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("上海"));
        put4.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000005"));
        put4.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("阿拉桑海人"));

        Put put5 = new Put("0006".getBytes());
        put5.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(6));
        put5.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小雨"));
        put5.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(22));
        put5.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("女"));
        put5.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("广东"));
        put5.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000006"));
        put5.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("洒洒水"));

        Put put6 = new Put("0007".getBytes());
        put6.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(7));
        put6.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("小瓜"));
        put6.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(22));
        put6.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("男"));
        put6.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("江苏"));
        put6.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15900000007"));
        put6.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("我是苏南的"));

        List<Put> listPut = new ArrayList<>();
        listPut.add(put);
        listPut.add(put2);
        listPut.add(put3);
        listPut.add(put4);
        listPut.add(put5);
        listPut.add(put6);

        table.put(listPut);
        table.close();
        System.out.println("插入多个数据成功");
    }


    public void getData(String tableName, String rowkey) throws IOException {

        //获取表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(rowkey.getBytes());
        //列族；列名
        get.addColumn("f2".getBytes(), "address".getBytes());

        //获取返回结果
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            //获取列族的名称
            String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            //获取列的名称
            String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            if (familyName.equals("f1") && columnName.equals("id") || columnName.equals("age")) {
                int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列族名： " + familyName + " ,列名： " + columnName + " ,列值：" + value);
            } else {
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列族名： " + familyName + " ,列名： " + columnName + " ,列值：" + value);
            }
        }
        //关闭表
        table.close();
    }

    public static void scanAllData(String tableName) throws IOException {

        //获取表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        //设置起始和结束的rowkey,扫描结果是：[)类型
        scan.setStartRow("0001".getBytes());
        scan.setStopRow("0008".getBytes());
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                //获取列族的名称
                String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                //获取列的名称
                String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                if (familyName.equals("f1") && columnName.equals("id") || columnName.equals("age")) {
                    int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列族名： " + familyName + " ,列名： " + columnName + " ,列值：" + value);
                } else {
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列族名： " + familyName + " ,列名： " + columnName + " ,列值：" + value);
                }
            }
        }
        //获取返回结果
        table.close();
    }

    public static void deleteData(String tableName) throws IOException {

        //获取表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete("0001".getBytes());
        table.delete(delete);
        table.close();

        System.out.println("删除成功");
    }

    public static void deleteTable(String tableName) throws IOException {

        //获取管理员对象
        Admin admin = connection.getAdmin();
        //禁用表
        admin.disableTable(TableName.valueOf(tableName));
        //删除表
        admin.deleteTable(TableName.valueOf(tableName));
        System.out.println("删除表成功");
    }
}